package com.storm.queue.core

import com.google.common.util.concurrent.ThreadFactoryBuilder
import com.storm.queue.NextTimeHolder
import com.storm.queue.TimeoutUtil
import com.storm.queue.model.Job
import com.storm.queue.register.AbstractTopicRegister
import com.storm.queue.thread.MoveJobThread
import com.storm.queue.thread.RetryTimeoutThread
import com.storm.queue.thread.ShutdownThread
import org.slf4j.LoggerFactory
import org.springframework.util.StringUtils
import java.util.concurrent.*
import java.util.concurrent.ThreadPoolExecutor.DiscardPolicy

class DelayQueueContext(private val delayOperation: DelayOperation) {

    /**TOPIC消费线程 标志位 */
    @Volatile
    private var topicThreadStop = false

    /**将异常情况下导致未被消费的Job 重新放回 待消费列表LIST中 */
    private val R_PUSH_NO_EXEC_JOB = ThreadPoolExecutor(
            0,
            50,
            30,
            TimeUnit.SECONDS,
            LinkedBlockingQueue(),
            ThreadFactoryBuilder().setDaemon(true).setNameFormat("异常未消费重入" + "-%d").build(),  //队列满了 丢掉任务不抛出异常;
            DiscardPolicy()
    )

    fun init() {
        //启动监听Bucket线程
        MoveJobThread.doMoveJob(delayOperation)
        //启动Topic的监听线程池
        runTopicsThreads()
        //启动每分钟唤醒一次线程
        runTimerNotify()
        //注册 优雅关机
        registerDestroy()
    }

    /**
     * 保证机制: 避免某台持有最小nextTime的机器挂掉;而导致不能准时消费
     * 唤醒线程: 每分钟唤醒一次  搬运线程
     */
    private val TIMER_NOTIFY = Executors.newScheduledThreadPool(1)

    /**redisDelayQueue: 接口的异步调用 线程池; */
    private val DelayQ_ASYNC = Executors.newCachedThreadPool()

    val log = LoggerFactory.getLogger(this.javaClass)
    //装载任务执行器的hash表
    private val topicRegisterHolder = HashMap<String, AbstractTopicRegister>()

    fun addTopicRegister(abstractTopicRegister: AbstractTopicRegister) {
        topicRegisterHolder[abstractTopicRegister.getTopic()] = abstractTopicRegister
    }
    /**
     * 启动消费TOPIC_LIST的线程
     * 每一个TOPIC对应一个线程池;
     * 每个线程池设置的 核心线程数都用阻塞原语BLPOP去阻塞获取LIST的元素;
     * 一个核心线程消费一个元素;不使用线程池的阻塞队列;避免服务器宕机之后这些任务丢失; 阻塞队列就是用的redis的List
     * @失败重试:
     * 1. 业务方的回调接口失败异常会重试
     * 默认重试2次,总共最多消费3次; 需要重试的Job会放在List的队尾中等待消费;
     * 业务方也可以设置  retryCount = -1 ;来阻止重试
     * 2. BLPOP成功,但是获取Job的时候连接超时或者异常
     * 这种不算回调接口异常,BLPOP已经删除了元素,所有需要重新放回到待消费列表
     */
    private fun runTopicsThreads() {
        //初始化Threads-topic
        log.info("初始化Topic线程runTopicsThreads topic.size:{}", topicRegisterHolder.size)
        for ((_, register) in topicRegisterHolder) {
            runTopicThreads(register)
        }
    }

    private fun runTopicThreads(register: AbstractTopicRegister) {
        //同时执行的线程数量不能超过register.getMaxPoolSize()数量,
        // 因为不打算将任务放入到线程池的阻塞队列,直接用redis的list当做阻塞队列,防止宕机丢失任务
        val semaphore = Semaphore(register.maxPoolSize)
        val registerThread = Thread(Runnable {
            log.info("创建Topic:{}线程;topicThreadStop:{}", register.getTopic(), topicThreadStop)
            while (!topicThreadStop) {
                try {
                    //可用许可证数量
                    val availablePermits = semaphore.availablePermits()
                    if (availablePermits == 0) {
                        //如果当前可用的许可不够 阻塞获取一个信号量;这里就是用来当做阻塞的功能
                        semaphore.acquire(1)
                        semaphore.release()
                    }
                    val topicIdR = delayOperation.queuePop(register.getTopic())

                    if (StringUtils.isEmpty(topicIdR)) {
                        Thread.sleep(7000)
                        continue
                    }
                    log.info("任务队列中获取的topic--{}", topicIdR)
                    //获取许可
                    semaphore.acquire()
                    val jobId = topicIdR!!.replace("\"".toRegex(), "")
                    register.getTOPIC_THREADS().execute {
                        log.info("执行线程操作")
                        var isFail = false
                        var job: Job? = null
                        try {
                            job = delayOperation.getJob(jobId)
                            if (job != null) {
                                try {
                                    checkTimeoutExecute(register.methodTimeout, register, job)
                                    if (job.retryCount> 0) {
                                        log.info("重试延迟任务第{}次重试消费成功,topicId:{},Job:{} ", job.retryCount, register.getTopic(), job.toString())

                                    } else {
                                        log.info("延迟任务消费成功,topicId:{},Job:{} ", register.getTopic(), job.toString())
                                    }
                                } catch (e: Exception) {
                                    if (job.retryCount > 0) {
                                        log.error("重试任务第{}次重试败,执行回调接口出错:topicId:{},Job:{},Err:{}", job.retryCount, register.getTopic(), job.toString(),e.message)
                                    } else {
                                        log.error("延迟任务消费失败,执行回调接口出错:topicId:{},Job:{},Err:{}", register.getTopic(), job.toString(), e.message)
                                    }
                                    isFail = true
                                    // 重试流程
                                    RetryTimeoutThread.callBackExceptionTryRetry(register, job, delayOperation)
                                }
                            }
                        } catch (e: Exception) {
                            /**topicId被Pop出来了,但是Job没有消费;都需要重新放入到队列中 */
                            /**topicId被Pop出来了,但是Job没有消费;都需要重新放入到队列中 */
                            if (job != null) {
                                log.error("延迟任务消费异常: TopicId已经Pop出来,但是Job没有被消费,重新RightPush进待消费列表;topicId:{},Err:{}",
                                        register.getTopic(), e.message)
                                againRightPush(register.getTopic(), job)
                            }
                        } finally {
                            /**1.执行成功 删除  2.就算执行失败,但是设置了不重试 也删除
                             * 3.就算执行失败,但是已经重试了2次 还是删除 */
                            /**1.执行成功 删除  2.就算执行失败,但是设置了不重试 也删除
                             * 3.就算执行失败,但是已经重试了2次 还是删除 */
                            if (job != null && (!isFail || job.retryCount > 2 || job.retryCount < 0)) {
                                //消费能力跟不上,这里可能会连接redis超时导致这里的Job没有被删除,但是影响不大
                                delayOperation.deleteJob(register.getTopic(), job.id)
                            }
                            semaphore.release()
                        }
                    }
                } catch (e: Exception) {
                    log.error(e.message, e)
                    try {
                        //避免redis断开连接一直刷屏,睡一秒
                        Thread.sleep(1000)
                    } catch (e1: InterruptedException) {
                    }
                }
            }
        })
        registerThread.isDaemon = true
        registerThread.name = register.getTopic() + "监听线程"
        registerThread.start()
    }

    @Throws(InterruptedException::class, ExecutionException::class, TimeoutException::class)
    private fun checkTimeoutExecute(timeout: Long, register: AbstractTopicRegister, job: Job) {
        TimeoutUtil.timeoutMethod(timeout,
                Callable{
                    register.execute(job)
                    false
                })
    }


    /**
     * 如果是redis连接超时;会出现这样的问题
     * 1.BLPOPKey 阻塞pop成功了； List中元素已经没有了
     * 2.通过这个pop出来的元素去 Jobs_Tables读取content;但是这个读取的时候连接超时了
     * 那么这个JOb就没有被消费了;所有这个时候要重新push到List中;
     */
    private fun againRightPush(topic: String, job: Job) {
        if (job.reentry > 2) {
            log.error("未被消费任务topicId:{},jobID:{}--重入了3次仍旧失败", topic, job.id)
        } else {
            R_PUSH_NO_EXEC_JOB.execute(Runnable {
                //获取JOb的方法连接失败了 异步重新放入;
                log.warn("未被消费任务topicId:{},jobId:{} 重新放入待消费队列", topic, job.id)
                job.reentry += 1
                delayOperation.retryJob(topic, job)
            })
        }
    }

    //注册优雅停机
    private fun registerDestroy() {
        Runtime.getRuntime().addShutdownHook(Thread(Runnable {
            log.info("延迟任务开始关机....")
            //关闭异步AddJob线程池
            ShutdownThread.closeExecutor(DelayQ_ASYNC, "异步AddJob线程池")
            //停止唤醒线程
            TIMER_NOTIFY.shutdown()
            //停止搬运线程
            MoveJobThread.toStop()
            //停止topic监听redis线程和 topic消费线程
            shutdownTopicThreads()
            //停止失败重试线程池
            RetryTimeoutThread.toStop()
            //停止 异常未消费JOb重入List线程池
            ShutdownThread.closeExecutor(R_PUSH_NO_EXEC_JOB, "异常未消费重入List线程池")
            log.info("延迟任务关机完毕....")
        }))
    }

    private fun shutdownTopicThreads() {
        topicThreadStop = true
        for ((key, value) in topicRegisterHolder) {
            ShutdownThread.closeExecutor(value.getTOPIC_THREADS(), key)
        }
    }

    /**
     * 每分钟触发一次搬运操作(这个只是宕机保证最坏情况延迟一分钟执行;)
     */
    private fun runTimerNotify() {
        TIMER_NOTIFY.scheduleWithFixedDelay(NextTimeHolder::setZeroAndNotify, 1, 1, TimeUnit.MINUTES)
    }

    //防止任务丢失，重新从主表读到时间桶
    fun setJobLast() {
        delayOperation.setJobLast()
    }
}

